package com.bang.study.batch;

import com.bang.study.constants.PublicConstants;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

/**
 * @Auther: Bang
 * @Date: 2020/5/6 22:39
 * @Description:
 */
public class Producer {

    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer(PublicConstants.TEST_PRODUCER_GROUP);
        producer.setNamesrvAddr(PublicConstants.NAME_ADDR);
        producer.start();
        List<Message> msgList = new ArrayList<>();
        Message msg1 = new Message("BATCH_TOPIC","tag1","我是消息1".getBytes());
        Message msg2 = new Message("BATCH_TOPIC","tag2","我是消息2".getBytes());
        Message msg3 = new Message("BATCH_TOPIC","tag3","我是消息3".getBytes());
        msgList.add(msg1);
        msgList.add(msg2);
        msgList.add(msg3);
        //把大的消息分裂成若干个小的消息
        ListSplitter splitter = new ListSplitter(msgList);
        while(splitter.hasNext()){
            try{
                List<Message>  listItem=splitter.next();
                producer.send(listItem);
            }catch(Exception e){
                e.printStackTrace();
                //处理error
            }
        }
        System.out.println("发送成功");
        producer.shutdown();
    }
}
